package com.shujia.flink.core;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

import java.time.Duration;

public class Demo5EventTime {
    public static void main(String[] args) throws Exception {

        /*
         * 事件时间：数据中有一个时间字段，使用数据的时间字段触发计算，代替真实的时间，可以反应数据真实发生的顺序，计算更有意义
         */

        /*
java,1685433130000
java,1685433131000
java,1685433132000
java,1685433134000
java,1685433135000
java,1685433137000
java,1685433139000
         */
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        /*
         *水位线对齐
         * 1、当上游有多个task时，下游task会取上游task水位线的最小值，如果数据量小。水位线就很难对齐，窗口就不会触发计算
         */

        env.setParallelism(1);

        DataStream<String> linesDS = env.socketTextStream("master", 8888);

        //解析数据
        DataStream<Tuple2<String, Long>> tsDS = linesDS.map(line -> {
            String[] split = line.split(",");
            String word = split[0];
            long ts = Long.parseLong(split[1]);
            return Tuple2.of(word, ts);
        }, Types.TUPLE(Types.STRING, Types.LONG));

        /*
         * 指定时间字段和水位线生成策略
         */
        DataStream<Tuple2<String, Long>> assDS = tsDS
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                //指定水位线生产策略，水位线等于最新一条数据的时间戳，如果数据乱序可能会丢失数据
                                //.<Tuple2<String, Long>>forMonotonousTimestamps()
                                //水位线前移时间（数据最大乱序时间）
                                .<Tuple2<String, Long>>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                                //指定时间字段
                                .withTimestampAssigner((event, ts) -> event.f1)
                );


        /*
         *每隔5秒统计单词的数量
         */
        DataStream<Tuple2<String, Integer>> kvDS = assDS
                .map(kv -> Tuple2.of(kv.f0, 1), Types.TUPLE(Types.STRING, Types.INT));

        KeyedStream<Tuple2<String, Integer>, String> keyByDS = kvDS
                .keyBy(kv -> kv.f0);

        //TumblingEventTimeWindows:滚动的事件时间窗口
        WindowedStream<Tuple2<String, Integer>, String, TimeWindow> windowDS = keyByDS
                .window(TumblingEventTimeWindows.of(Time.seconds(5)));

        windowDS.sum(1).print();

        env.execute();

    }
}
